package bigo;

import bigo.data.ArrayVector;
import bigo.data.MatrixMeta;
import bigo.data.RowIndex;
import bigo.data.Vector;
import bigo.data.Vector.Wrapper;
import bigo.lib.AgreegationDriver;
import bigo.lib.InputFormat;
import bigo.lib.InputMatrix;
import bigo.lib.OutputFormat;
import bigo.lib.OutputMatrix;
import bigo.lib.Reducer;
import bigo.lib.TextMatrixInputFormat;
import bigo.lib.TextMatrixOutputFormat;
import java.io.IOException;
import java.util.HashMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ToolRunner;

/**
 *
 * @author songliu
 */
public class SVMAggregationDriver extends AgreegationDriver {

    @InputMatrix
    public static MatrixMeta ETA, Y, Z, ALPHA;
    @OutputMatrix
    public static MatrixMeta O;
    @InputFormat
    public static Class inputFormat = TextMatrixInputFormat.class;
    @OutputFormat
    public static Class OutputFormat = TextMatrixOutputFormat.class;

    public SVMAggregationDriver() {
    }

    public SVMAggregationDriver(String[] args) {
        String etaPath = args[0];
        String alphaPath = args[1];
        String YPath = args[2];
        String ZPath = args[3];
        String oPath = args[4];

        int m = Integer.parseInt(args[5]), n = Integer.parseInt(args[6]);

        SVMAggregationDriver.ETA = new MatrixMeta(etaPath, m, n);
        SVMAggregationDriver.ALPHA = new MatrixMeta(alphaPath, m, n);
        SVMAggregationDriver.Y = new MatrixMeta(YPath, m, n);
        SVMAggregationDriver.Z = new MatrixMeta(ZPath, m, n);
        SVMAggregationDriver.O = new MatrixMeta(oPath, m, n);
    }

    public SVMAggregationDriver(MatrixMeta ETA,
            MatrixMeta Y,
            MatrixMeta Z,
            MatrixMeta ALPHA,
            MatrixMeta O) {
        SVMAggregationDriver.O = O;
        SVMAggregationDriver.ALPHA = ALPHA;
        SVMAggregationDriver.ETA = ETA;
        SVMAggregationDriver.Y = Y;
        SVMAggregationDriver.Z = Z;
    }

    @bigo.lib.Mapper
    public static class SVMMapper extends AgreegationMapper {
    }

    public static Vector agreegate(RowIndex index, HashMap<String, Vector> vectors) {
        Vector y = vectors.get(Y.path);
        Vector eta = vectors.get(ETA.path);
        Vector z = vectors.get(Z.path);
        Vector alpha = vectors.get(ALPHA.path);

        for (Integer i : eta.keySet()) {
            float t = Math.abs(alpha.get(i)) + SVMReducer.eta * (1 - y.get(i) * z.get(i));
            if (t > 0) {
                alpha.put(i, y.get(i) * t);
            } else {
                alpha.put(i, .0f);
            }
        }
        return alpha;
    }

    @Reducer
    public static class SVMReducer extends AgreegationReducer {

        private static float eta;
        @Override
        public bigo.data.Vector agreegate(RowIndex index, HashMap<String, bigo.data.Vector> vectors) {
            return SVMAggregationDriver.agreegate(index, vectors);
        }

        @Override
        protected void setup(
                org.apache.hadoop.mapreduce.Reducer.Context context)
                throws IOException,
                InterruptedException {
            super.setup(context);
            eta = context.getConfiguration().getFloat("eta.rate", 0.2f);
        }


    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Configuration(),
                new SVMAggregationDriver(args),
                args);
    }
}
